package com.heima.article.stream;

import com.alibaba.fastjson.JSON;
import com.heima.common.constants.HotArticleConstants;
import com.heima.model.article.mess.ArticleVisitStreamMess;
import com.heima.model.mess.UpdateArticleMess;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.Aggregator;
import org.apache.kafka.streams.kstream.Initializer;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.springframework.context.annotation.Bean;
import org.springframework.stereotype.Component;

import java.time.Duration;

/**
 * kafkastream监听用户行为服务发送的数据,统一进行流式计算
 */
@Slf4j
@Component
public class HotArticleStreamHandler {


    @Bean
    public KStream<String, String> kStream(StreamsBuilder streamsBuilder) {
        //创建kstream对象 ,指定从哪个topic中接收消息
        KStream<String, String> stream = streamsBuilder.stream(HotArticleConstants.HOT_ARTICLE_SCORE_TOPIC);

        stream.map((key, value) -> {  //接收到的value格式为: {"articleId":"12345","type":"COLLECTION","add":"1"}
                    UpdateArticleMess mess = JSON.parseObject(value, UpdateArticleMess.class);
                    //key:文章id1232444  key: likes:1
                    return new KeyValue<>(mess.getArticleId().toString(), mess.getType() + ":" + mess.getAdd());
                })
                //按照文章id进行聚合
                .groupBy((key, value) -> key)
                //时间窗口
                .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
                //自行完成聚合计算
                .aggregate(new Initializer<String>() {
                    /**
                     * 初始化方法, 每个时间窗口内进行聚合前都行执行一次,也只执行一次
                     * @return
                     */
                    @Override
                    public String apply() {
                        return "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0";
                    }
                }, new Aggregator<String, String, String>() {
                    /**
                     * 对stream中每个消息聚合处理时都会执行一次该方法
                     * applyValue代表 初始方法中apply()方法的返回值,也是时间窗口内计算的的返回值
                     *
                     *真正的聚合操作: 返回值是消息的value
                     */
                    @Override
                    public String apply(String key, String value, String applyValue) {
                        log.info("<<<<<<<<>>>>>>>>>applyValue: {}", applyValue);
                        if (StringUtils.isBlank(value)) {
                            return applyValue;
                        }
                        /**
                         * 获取初始值,也是时间窗口内计算之后的值
                         *  "COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
                         */
                        int col = 0, com = 0, lik = 0, vie = 0;

                        String[] applyValueAry = applyValue.split(",");
                        for (String applyVal : applyValueAry) {
                            String[] split = applyVal.split(":");
                            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                                case COLLECTION:
                                    col = Integer.valueOf(split[1]);
                                    break;
                                case COMMENT:
                                    com = Integer.valueOf(split[1]);
                                    break;
                                case LIKES:
                                    lik = Integer.valueOf(split[1]);
                                    break;
                                case VIEWS:
                                    vie = Integer.valueOf(split[1]);
                                    break;
                            }
                        }
                        /**
                         * 累加操作
                         */
                        //value:  likes:1
                        String[] valueAry = value.split(":");
                        switch (UpdateArticleMess.UpdateArticleType.valueOf(valueAry[0])) {
                            case COLLECTION:
                                col += Integer.valueOf(valueAry[1]);
                                break;
                            case COMMENT:
                                com += Integer.valueOf(valueAry[1]);
                                break;
                            case LIKES:
                                lik += Integer.valueOf(valueAry[1]);
                                break;
                            case VIEWS:
                                vie += Integer.valueOf(valueAry[1]);
                                break;
                        }
                        String formatStr = String.format("COLLECTION:%d,COMMENT:%d,LIKES:%d,VIEWS:%d", col, com, lik, vie);
                        log.info("当前文章id:{}", key);
                        log.info("当前时间窗口内消息的处理结果为:{}", formatStr);
                        return formatStr;
                    }
                }).toStream()
                .map((key, value) -> {
                    //value: COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0
                    return new KeyValue<>(key.key().toString(), formatObj(key.key().toString(), value.toString()));
                })
                .to(HotArticleConstants.HOT_ARTICLE_INCR_HANDLE_TOPIC);

        return stream;
    }

    /**
     * 格式化消息的value数据 (将数据格式化为ArticleVisitStreamMess对象)
     *
     * @param articleId
     * @param value     :"COLLECTION:0,COMMENT:0,LIKES:0,VIEWS:0"
     * @return
     */
    private String formatObj(String articleId, String value) {
        ArticleVisitStreamMess mess = new ArticleVisitStreamMess();
        mess.setArticleId(Long.valueOf(articleId));

        String[] valueAry = value.split(",");
        for (String val : valueAry) {
            String[] split = val.split(":");
            switch (UpdateArticleMess.UpdateArticleType.valueOf(split[0])) {
                case COLLECTION:
                    mess.setCollect(Integer.valueOf(split[1]));
                    break;
                case COMMENT:
                    mess.setComment(Integer.valueOf(split[1]));
                    break;
                case LIKES:
                    mess.setLike(Integer.valueOf(split[1]));
                    break;
                case VIEWS:
                    mess.setView(Integer.valueOf(split[1]));
                    break;
            }
        }
        return JSON.toJSONString(mess);
    }

}
